package com.demo.spring.utils;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;

/**
 * HDFS操作类
 * @author admin
 *
 */
public class HDFSUtil {

	public static final Log log = LogFactory.getLog(HDFSUtil.class);
	
	private HDFSUtil() {
		
	}

	/**
	 * 判断路径是否存在
	 * 
	 * @param conf
	 * @param path
	 * @return
	 * @throws IOException
	 */
	public static boolean exits(FileSystem fs, String path) throws IOException {
		return fs.exists(new Path(path));
	}

	/**
	 * 创建文件
	 * 
	 * @param conf
	 * @param filePath
	 * @param contents
	 * @throws IOException
	 */
	public static boolean createFile(FileSystem fs, String filePath, byte[] contents)
			throws IOException {
		Path path = new Path(filePath);
		FSDataOutputStream outputStream = fs.create(path);
		outputStream.write(contents);
		outputStream.close();
		log.debug(filePath+" be saved~");
		return true;
	}

	/**
	 * 创建文件
	 * 
	 * @param conf
	 * @param filePath
	 * @param fileContent
	 * @throws IOException
	 */
	public static boolean createFile(FileSystem fs, String filePath, String fileContent)
			throws IOException {
		return createFile(fs, filePath, fileContent.getBytes());
	}

	/**
	 * @param conf
	 * @param localFilePath
	 * @param remoteFilePath
	 * @throws IOException
	 */
	public static void copyFromLocalFile(FileSystem fs, String localFilePath,
			String remoteFilePath) throws IOException {
		Path localPath = new Path(localFilePath);
		Path remotePath = new Path(remoteFilePath);
		fs.copyFromLocalFile(false, true, localPath, remotePath);
	}

	/**
	 * 删除目录或文件
	 * 
	 * @param conf
	 * @param remoteFilePath
	 * @param recursive
	 * @return
	 * @throws IOException
	 */
	public static boolean deleteFile(FileSystem fs, String remoteFilePath, boolean recursive)
			throws IOException {
		boolean result = fs.delete(new Path(remoteFilePath), recursive);
		return result;
	}

	/**
	 * 文件重命名
	 * 
	 * @param conf
	 * @param oldFileName
	 * @param newFileName
	 * @return
	 * @throws IOException
	 */
	public static boolean renameFile(FileSystem fs, String oldFileName, String newFileName)
			throws IOException {
		Path oldPath = new Path(oldFileName);
		Path newPath = new Path(newFileName);
		boolean result = fs.rename(oldPath, newPath);
		return result;
	}

	/**
	 * 创建目录
	 * 
	 * @param conf
	 * @param dirName
	 * @return
	 * @throws IOException
	 */
	public static boolean createDirectory(FileSystem fs, String dirName) throws IOException {
		Path dir = new Path(dirName);
		boolean result = false;
		if (!fs.exists(dir)) {
			result = fs.mkdirs(dir);
		}
		return result;
	}

	/**
	 * 列出指定路径下的所有文件(不包含目录)
	 * 
	 * @param conf
	 * @param basePath
	 * @param recursive
	 */
	public static RemoteIterator<LocatedFileStatus> listFiles(FileSystem fs, String basePath,
			boolean recursive) throws IOException {
		RemoteIterator<LocatedFileStatus> fileStatusRemoteIterator = fs
				.listFiles(new Path(basePath), recursive);
		return fileStatusRemoteIterator;
	}

	/**
	 * 列出指定路径下的文件（非递归）
	 * 
	 * @param conf
	 * @param basePath
	 * @return
	 * @throws IOException
	 */
	public static RemoteIterator<LocatedFileStatus> listFiles(FileSystem fs, String basePath)
			throws IOException {
		RemoteIterator<LocatedFileStatus> remoteIterator = fs.listFiles(
				new Path(basePath), false);
		return remoteIterator;
	}

	/**
	 * 列出指定目录下的文件\子目录信息（非递归）
	 * 
	 * @param conf
	 * @param dirPath
	 * @return
	 * @throws IOException
	 */
	public static FileStatus[] listStatus(FileSystem fs, String dirPath) throws IOException {
		FileStatus[] fileStatuses = fs.listStatus(new Path(dirPath));
		return fileStatuses;
	}

	/**
	 * 读取文件内容
	 * 
	 * @param conf
	 * @param filePath
	 * @return
	 * @throws IOException
	 */
	public static byte[] readFile(FileSystem fs, String filePath) throws IOException {
		byte[] fileContent = null;
		Path path = new Path(filePath);
		if (fs.exists(path)) {
			InputStream inputStream = null;
			ByteArrayOutputStream outputStream = null;
			try {
				inputStream = fs.open(path);
				outputStream = new ByteArrayOutputStream(
						inputStream.available());
				IOUtils.copyBytes(inputStream, outputStream, fs.getConf());
				fileContent = outputStream.toByteArray();
			} finally {
				IOUtils.closeStream(inputStream);
				IOUtils.closeStream(outputStream);
			}
		}
		return fileContent;
	}

	/**
	 * 下载 hdfs上的文件
	 * 
	 * @param conf
	 * @param uri
	 * @param remote
	 * @param local
	 * @throws IOException
	 */
	public static void download(FileSystem fs, String remote, String local) throws IOException {
		Path path = new Path(remote);
		fs.copyToLocalFile(path, new Path(local));
		log.debug("download: from" + remote + " to " + local);
	}
}